// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: linchuan

#include "coin2/exchange/bitstamp/order_v2/parser.h"

#include <memory>
#include <set>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/replace.hpp>
#include <rapidjson/document.h>

#include "coin/proto/coin_query.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/util/json_util.h"
#include "quickfix/Message.h"

namespace coin2::exchange::bitstamp::order_v2 {

namespace {

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductFillElement;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;

using OrderContext = base::order::IOrderContext;
using symbology::BitstampCurrency;
using symbology::BitstampProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
using OrderEventType = coin::proto::OrderEvent::OrderEventType;

inline int64_t GetOrderTimestamp(const JsonValue& data) {
  std::string time_str = data["datetime"].GetString();
  if (time_str.length() == 19) {
    // from "2019-11-25 11:17:54" to 2019-11-25T11:17:54.000Z
    time_str.replace(10, 1, "T");
    return TimestampFromIso8601(time_str + ".000Z");
  } else {
    // from "2019-11-25 11:17:54.441231" to 2019-11-25T11:17:54.441Z
    time_str.replace(time_str.begin() + 23, time_str.end(), "Z");
    time_str.replace(10, 1, "T");
    return TimestampFromIso8601(time_str);
  }
}

inline coin::proto::OrderSide GetOrderSide(const std::string_view& side) {
  if (side == "0") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "1") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline bool IsBitstampErrMessage(const JsonValue& doc) {
  return doc.HasMember("reason") || doc.HasMember("error");
}

inline std::string GetBitstampExchangeOrderErrorMsg(const JsonValue& json) {
  if (json.HasMember("reason")) {
    if (json["reason"].IsObject()) {
      return JsonToString(json["reason"]);
    } else {
      return json["reason"].GetString();
    }
  }
  return json["error"].GetString();
}

inline coin::proto::OrderErrorCode GetOrderErrorCodeFromBitstamp(const JsonValue& json) {
  using coin::proto::OrderErrorCode;
  auto result = coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR;
  std::string error_code = GetBitstampExchangeOrderErrorMsg(json);

  if (error_code == "0") {
    result = OrderErrorCode::ORDER_NO_ERROR;
  } else if (error_code.find("Check your account balance for details") != std::string::npos) {
    result = OrderErrorCode::SUBMIT_ORDER_INSUFFICIENT_FUND;
  } else if (error_code == "Order not found") {
    result = OrderErrorCode::ORDER_NOT_FOUND;
  } else {
    LOG(ERROR) << "Unknown error code: " << error_code;
  }

  return result;
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!json.HasMember("channel")) {
    return false;
  }

  if (!json.HasMember("event") || json["event"] == "bts:subscription_succeeded") {
    return false;
  }

  std::string_view ws_type = json["channel"].GetString();
  return ws_type.compare(0, 5, "live_") == 0;
}

}  // namespace

void BitstampParser::ParseRestAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance BitstampParser::ParseBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Bitstamp");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeLower()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

bool BitstampParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "bch_available": "0.00000000",
    "bch_balance": "0.00000000",
    "bch_reserved": "0.00000000",
    "bchbtc_fee": "0.25",
    "bcheur_fee": "0.25",
    "bchusd_fee": "0.25",
    "btc_available": "0.00000000",
    "btc_balance": "0.00000000",
    "btc_reserved": "0.00000000",
    "btceur_fee": "0.25",
    "btcusd_fee": "0.25",
    "eth_available": "0.00000000",
    "eth_balance": "0.00000000",
    "eth_reserved": "0.00000000",
    "ethbtc_fee": "0.25",
    "etheur_fee": "0.25",
    "ethusd_fee": "0.25",
    "eur_available": "0.00",
    "eur_balance": "0.00",
    "eur_reserved": "0.00",
    "eurusd_fee": "0.25",
    "ltc_available": "0.00000000",
    "ltc_balance": "0.00000000",
    "ltc_reserved": "0.00000000",
    "ltcbtc_fee": "0.25",
    "ltceur_fee": "0.25",
    "ltcusd_fee": "0.25",
    "usd_available": "0.00",
    "usd_balance": "0.00",
    "usd_reserved": "0.00",
    "xrp_available": "0.00000000",
    "xrp_balance": "0.00000000",
    "xrp_reserved": "0.00000000",
    "xrpbtc_fee": "0.25",
    "xrpeur_fee": "0.25",
    "xrpusd_fee": "0.25"
  }
  */
  if (doc.IsObject() && IsBitstampErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }

  std::optional<double> last_available;
  for (const auto& info : doc.GetObject()) {
    std::string name = info.name.GetString();
    const auto& data = info.value.GetString();
    if (name.compare(name.length() - 9, 9, "available") == 0) {
      last_available = stod(data);
      continue;
    }
    if (name.compare(name.length() - 7, 7, "balance") == 0 && last_available) {
      CurrencyBalance currency_balance;
      currency_balance.set_currency_native(name.substr(0, name.length() - 8));
      double current_free = *last_available;
      double current_total = stod(data);
      currency_balance.set_available(current_free);
      currency_balance.set_total(current_total);
      currency_balance.set_hold(current_total - current_free);
      raw_balance_map->emplace(currency_balance.currency_native(), currency_balance);
    }
  }

  return true;
}

ProductOrderElement BitstampParser::ParseOrderInfo(const JsonValue& data) {
  ProductOrderElement order;
  order.set_market_type(MarketType::Spot);
  order.set_exchange_type(ExchangeType::Bitstamp);
  order.set_api_version("v2");
  order.set_exchange_order_id(data["id"].GetString());
  std::string symbol_raw = data["currency_pair"].GetString();
  boost::replace_all(symbol_raw, "/", "");
  order.set_symbol(StringToLowerCopy(symbol_raw));
  order.set_price(stod(data["price"].GetString()));
  order.set_qty(stod(data["amount"].GetString()));
  order.set_side(GetOrderSide(data["type"].GetString()));
  order.set_timestamp(msg_->timestamp);
  order.set_is_live(true);
  return order;
}

void BitstampParser::ParseRestOpensInfo() {
  /*
  [
  {
    "price": "169.54",
    "currency_pair": "ETH/USD",
    "datetime": "2019-04-24 03:25:31",
    "amount": "0.10000000",
    "type": "1",
    "id": "3174363621"
  }
  ]
  */
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOrdersInfo] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsBitstampErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOpensInfo] " << JsonToString(data);
    return;
  }

  if (!data.IsArray()) {
    LOG(ERROR) << "[ParseRestOpensInfo] Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (auto& info : data.GetArray()) {
    auto order = product_order_bundle->add_each_order();
    auto new_order = ParseOrderInfo(info);
    order->MergeFrom(new_order);
    product_order_bundle_found = true;
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

ProductOrderBundle BitstampParser::ParseWsOrderInfoListV2(const JsonValue& data) {
  /*
  {
    "data":{
      "id":1238002162155520,
      "id_str":"1238002162155520",
      "order_type":0,
      "datetime":"1591081596",
      "microtimestamp":"1591081596277000",
      "amount":0.013,                           // linchuan: this means qty left
      "amount_str":"0.01300000",
      "price":10074.96,                         // linchuan: become filled price if been filled
      "price_str":"10074.96"
    },
    "channel":"live_orders_btcusd",
    "event":"order_created"
  }
  */
  ProductOrderBundle product_order_bundle;
  std::string order_id_str = data["data"]["id"].GetString();
  auto* oc = GetOcm()->FindByExchangeOrderId(order_id_str);
  if (oc == nullptr) {
    return product_order_bundle;
  }
  DLOG(INFO) << "my ws order msg: " << JsonToString(data);
  auto order = product_order_bundle.add_each_order();
  order->set_market_type(MarketType::Spot);
  order->set_exchange_type(ExchangeType::Bitstamp);
  order->set_api_version("v2");
  order->set_symbol(oc->native_product());
  order->set_exchange_order_id(order_id_str);
  order->set_side(GetOrderSide(data["data"]["order_type"].GetString()));
  order->set_timestamp(msg_->timestamp);
  std::string_view status = data["event"].GetString();
  if (status == "order_created") {
    order->set_is_live(true);
  } else if (status == "order_changed") {  // partially filled
    order->set_is_live(true);
  } else if (status == "order_deleted") {
    double qty_left = stod(data["data"]["amount"].GetString());
    if (qty_left < 1e-8) {
      order->set_fully_filled(true);
    } else {
      order->set_cancel_confirmed(true);
    }
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
  return product_order_bundle;
}

void BitstampParser::ParseRestFillInfoList() {
  /*
  [
    {
      "usd": "344.86",
      "btc_usd": 6897.30000000,
      "order_id": 4377817675,
      "datetime": "2019-11-25 11:17:54.441231",
      "fee": "0.18000000",
      "btc": "-0.05000000",
      "type": "2",
      "id": 100966590,
      "eur": 0.0
    }
  ]
  */
  JsonDocument data_arr;
  if (!PrepareJsonDocument(&data_arr)) {
    LOG(ERROR) << "[ParseRestFillOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data_arr.IsObject() && IsBitstampErrMessage(data_arr)) {
    LOG(ERROR) << "[ParseRestFillInfoList] " << JsonToString(data_arr);
    return;
  }

  if (!data_arr.IsArray()) {
    LOG(ERROR) << "[ParseRestFillInfoList] Unexpected fill list format!" << JsonToString(data_arr);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  bool product_fill_bundle_found = false;
  for (auto& data : data_arr.GetArray()) {
    if (std::string_view(data["type"].GetString()) != "2") {
      continue;
    }
    std::string symbol;
    std::string base_currency;
    std::string native_symbol;
    for (const auto& info : data.GetObject()) {
      symbol = info.name.GetString();
      if (symbol != "order_id" && symbol.find("_") != std::string::npos) {
        std::vector<std::string> splitted;
        splitted = Split(symbol, "_");
        base_currency = splitted[0];
        native_symbol = splitted[0] + splitted[1];
        break;
      }
    }
    // ignore the fills long ago
    int64_t fill_timestamp_et = GetOrderTimestamp(data);
    int64_t fill_timestamp_ft = msg_->timestamp;
    if (fill_timestamp_ft - fill_timestamp_et > 100 * 1e9) {
      continue;
    }
    auto fill = product_fill_bundle->add_each_fill();

    std::string order_id = data["order_id"].GetString();
    fill->set_symbol(native_symbol);
    fill->set_fill_timestamp(fill_timestamp_ft);
    fill->set_exchange_order_id(order_id);
    fill->set_price(stod(data[symbol.c_str()].GetString()));
    double qty = stod(data[base_currency.c_str()].GetString());
    fill->set_qty(std::fabs(qty));
    fill->set_fill_id(data["id"].GetString());
    fill->set_fee(stod(data["fee"].GetString()));
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

ProductFillBundle BitstampParser::ParseWsFillInfoListV2(const JsonValue& data) {
  /*
  {
    "data": {
      "buy_order_id": 1238005052661760,
      "amount_str": "0.01054579",
      "timestamp": "1591082301",
      "microtimestamp": "1591082301967000",
      "id": 114762156,
      "amount": 0.01054579,
      "sell_order_id": 1238005037490177,
      "price_str": "10103.54",
      "type": 0,
      "price": 10103.54
    },
    "event": "trade",
    "channel": "live_trades"
  }
  */
  ProductFillBundle product_fill_bundle;
  std::vector<std::string> order_vec;
  order_vec.emplace_back(data["data"]["buy_order_id"].GetString());
  order_vec.emplace_back(data["data"]["sell_order_id"].GetString());

  for (auto& order_id : order_vec) {
    int64_t fill_timestamp_et = std::stoll(data["data"]["microtimestamp"].GetString()) * 1000;
    int64_t fill_timestamp_ft = msg_->timestamp;
    if (fill_timestamp_ft - fill_timestamp_et > 100 * 1e9) {
      continue;
    }
    auto* oc = GetOcm()->FindByExchangeOrderId(order_id);
    if (oc == nullptr) {
      continue;
    }
    DLOG(INFO) << "my ws fill msg: " << JsonToString(data);
    auto fill = product_fill_bundle.add_each_fill();
    BitstampProduct product =
        BitstampProduct::FromStr(oc->order_spec().product, oc->order_spec().order_created_time);
    fill->set_symbol(product.order_native_symbol());
    fill->set_fill_timestamp(fill_timestamp_ft);
    fill->set_exchange_order_id(order_id);
    fill->set_price(stod(data["data"]["price"].GetString()));
    fill->set_qty(stod(data["data"]["amount"].GetString()));
    fill->set_fill_id(data["data"]["id"].GetString());
  }
  return product_fill_bundle;
}

void BitstampParser::ParseSubmitResponse() {
  /*
  {
    "price": "9895.98",
    "amount": "0.00300000",
    "type": "1",
    "id": "1239036527185921",
    "datetime": "2020-06-05 05:15:26.804908"
  }
  */
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[BitstampSubmitOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[BitstampSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[BitstampSubmitOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (IsBitstampErrMessage(json)) {
    auto error_code = GetOrderErrorCodeFromBitstamp(json);
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(GetBitstampExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[BitstampSubmitOrder] Error Response" << http_context->res;
    }
    resp->set_success(false);
    return;
  }

  if (!json.HasMember("id")) {
    LOG(ERROR) << "[BitstampSubmitOrder] exchange order id not found\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR);
    resp->set_success(false);
    return;
  }
  resp->set_success(true);
  resp->set_exchange_order_id(json["id"].GetString());
}

void BitstampParser::ParseCancelResponse() {
  /*
  {
    "price": 9895.98,
    "amount": 0.003,
    "type": 1,
    "id": 1239036527185921
  }
  */
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[BitstampCancelOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[BitstampCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[BitstampCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (IsBitstampErrMessage(json)) {
    auto error_code = GetOrderErrorCodeFromBitstamp(json);
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(GetBitstampExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[BitstampCancelOrder] Error Response" << http_context->res;
    }
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_cancel_confirmed(true);
}

void BitstampParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (IsOrderUpdate(document)) {
    std::string ws_type = document["channel"].GetString();
    if (ws_type.compare(0, 11, "live_trades") == 0) {
      *result_.mutable_fill_list() = ParseWsFillInfoListV2(document);
      result_.set_type(MsgType::ORDER_FILL);
    } else {
      *result_.mutable_order_list() = ParseWsOrderInfoListV2(document);
      result_.set_type(MsgType::ORDER_STATUS);
    }
  }
}

void BitstampParser::ParseFixMessage() {
  if (!msg_ || msg_->fix_message == nullptr) {
    return;
  }
  const FIX::Message& message = *(msg_->fix_message);

  DLOG(INFO) << message;

  FIX::MsgType msg_type;
  if (!message.getHeader().getFieldIfSet(msg_type)) {
    return;
  }
  if (MSG_TYPE_SESSION_REJECT == msg_type) {
    // TODO(daniel) - session reject
  } else if (MSG_TYPE_EXECUTION_REPORT == msg_type) {
    ParseFixSubmitResponse(message);
    ParseFixCancelResponse(message);
    ParseFixFillInfo(message);
    ParseFixOrderInfo(message);
  } else if (MSG_TYPE_CANCEL_REJECT == msg_type) {
    ParseFixCancelResponse(message);
  } else if (MSG_TYPE_BUSINESS_REJECT == msg_type) {
    // TODO(daniel) - business reject
  }
}

void BitstampParser::ParseFixSubmitResponse(const FIX::Message& message) {
  FIX::ExecType exec_type;
  if (!message.getFieldIfSet(exec_type)) {
    return;
  }
  if (EXEC_TYPE_ORDER_REJECTED != exec_type) {
    return;
  }
  std::optional<int64_t> proc_order_id;
  FIX::ClOrdID client_id;
  if (message.getFieldIfSet(client_id)) {
    proc_order_id = ToInt64(client_id.getValue());
  }
  if (!proc_order_id) {
    LOG(ERROR) << "proce_order_id not found. " << message;
    return;
  }
  FIX::Symbol symbol;
  message.getFieldIfSet(symbol);

  SubmitOrderResponse& resp = *result_.mutable_submit_response();
  resp.Clear();
  resp.set_symbol(GetLocaleSymbol(symbol.getValue()));
  resp.set_proc_order_id(*proc_order_id);
  resp.set_timestamp(msg_->timestamp);
  resp.set_error_code(coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR);
  FIX::OrdRejReason error_code;
  if (message.getFieldIfSet(error_code)) {
    // TODO(daniel) - confirm error code
    if (3 == error_code) {
      resp.set_error_code(coin::proto::OrderErrorCode::SUBMIT_ORDER_INSUFFICIENT_FUND);
    }
  }
  FIX::Text error_reason;
  if (message.getFieldIfSet(error_reason)) {
    resp.set_exchange_error_msg(error_reason.getValue());
  }
  resp.set_success(false);
  result_.set_type(MsgType::SUBMIT_RESPONSE);
}

void BitstampParser::ParseFixCancelResponse(const FIX::Message& message) {
  FIX::MsgType msg_type;
  if (!message.getHeader().getFieldIfSet(msg_type)) {
    return;
  }
  if (MSG_TYPE_CANCEL_REJECT != msg_type) {
    return;
  }
  std::optional<int64_t> proc_order_id;
  FIX::ClOrdID client_id;
  if (message.getFieldIfSet(client_id)) {
    proc_order_id = ToInt64(client_id.getValue());
  }
  if (!proc_order_id) {
    LOG(ERROR) << "proc_order_id not found. " << message;
    return;
  }
  FIX::OrderID order_id;
  message.getFieldIfSet(order_id);

  CancelOrderResponse& resp = *result_.mutable_cancel_response();
  resp.Clear();
  resp.set_proc_order_id(*proc_order_id);
  resp.set_exchange_order_id(order_id.getValue());
  resp.set_timestamp(msg_->timestamp);
  resp.set_error_code(coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR);
  FIX::CxlRejReason error_code;
  if (message.getFieldIfSet(error_code)) {
    if (1 == error_code) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NOT_FOUND);
    }
  }
  FIX::Text error_reason;
  if (message.getFieldIfSet(error_reason)) {
    resp.set_exchange_error_msg(error_reason.getValue());
  }
  resp.set_success(false);
  result_.set_type(MsgType::CANCEL_RESPONSE);
}

void BitstampParser::ParseFixFillInfo(const FIX::Message& message) {
  FIX::ExecType exec_type;
  if (!message.getFieldIfSet(exec_type)) {
    return;
  }
  if (EXEC_TYPE_ORDER_FILL != exec_type) {
    return;
  }
  std::optional<int64_t> proc_order_id;
  FIX::ClOrdID client_id;
  if (message.getFieldIfSet(client_id)) {
    proc_order_id = ToInt64(client_id.getValue());
  }
  if (!proc_order_id) {
    LOG(ERROR) << "proc_order_id not found. " << message;
    return;
  }
  FIX::OrderID order_id;
  FIX::Symbol symbol;
  FIX::LastPx price;
  FIX::LastQty order_qty;
  FIX::ExecID exec_id;

  message.getFieldIfSet(order_id);
  message.getFieldIfSet(symbol);
  message.getFieldIfSet(price);
  message.getFieldIfSet(order_qty);
  message.getFieldIfSet(exec_id);
  if (auto oc = GetOcm()->FindByExchangeOrderId(order_id.getValue()); !oc) {
    return;
  }
  ProductFillBundle& product_fill_bundle = *result_.mutable_fill_list();
  product_fill_bundle.Clear();
  auto fill = product_fill_bundle.add_each_fill();
  fill->Clear();
  fill->set_symbol(GetLocaleSymbol(symbol.getValue()));
  fill->set_fill_timestamp(msg_->timestamp);
  fill->set_exchange_order_id(order_id.getValue());
  fill->set_proc_order_id(*proc_order_id);
  fill->set_price(price.getValue());
  fill->set_qty(order_qty.getValue());
  fill->set_fill_id(ExtractFillID(exec_id.getValue()));
  result_.set_type(MsgType::ORDER_FILL);
}

void BitstampParser::ParseFixOrderInfo(const FIX::Message& message) {
  FIX::ExecType exec_type;
  if (!message.getFieldIfSet(exec_type)) {
    return;
  }

  if (ORD_STATUS_NEW_ORDER == exec_type || EXEC_TYPE_ORDER_CANCELLED == exec_type ||
      EXEC_TYPE_ORDER_FILL == exec_type) {
    FIX::OrderID order_id;
    FIX::Symbol symbol;
    FIX::LeavesQty order_qty;
    FIX::ClOrdID client_id;
    FIX::Side side;
    FIX::OrdType order_type;
    FIX::OrdStatus order_status;
    message.getFieldIfSet(order_id);
    message.getFieldIfSet(symbol);
    message.getFieldIfSet(order_qty);
    message.getFieldIfSet(client_id);
    message.getFieldIfSet(side);
    message.getFieldIfSet(order_type);
    message.getFieldIfSet(order_status);
    auto proc_order_id = ToInt64(client_id.getValue());
    if (!proc_order_id) {
      LOG(ERROR) << "proc_order_id not found. " << message;
      return;
    }

    ProductOrderBundle& product_order_bundle = *result_.mutable_order_list();
    product_order_bundle.Clear();
    auto order = product_order_bundle.add_each_order();
    order->Clear();
    order->set_market_type(MarketType::Spot);
    order->set_exchange_type(ExchangeType::Bitstamp);
    order->set_api_version("v2");
    order->set_exchange_order_id(order_id.getValue());
    order->set_symbol(GetLocaleSymbol(symbol.getValue()));
    order->set_qty(order_qty.getValue());
    order->set_proc_order_id(*proc_order_id);
    if ('1' == side.getValue()) {
      order->set_side(coin::proto::OrderSide::BUY_ORDER);
    } else if ('2' == side.getValue()) {
      order->set_side(coin::proto::OrderSide::SELL_ORDER);
    }
    order->set_timestamp(msg_->timestamp);
    if (ORD_STATUS_NEW_ORDER == order_status || ORD_STATUS_PARTIALLY_FILLED == order_status) {
      order->set_is_live(true);
    } else if (ORD_STATUS_FULLY_FILLED == order_status) {
      order->set_fully_filled(true);
    } else if (ORD_STATUS_CANCELLED == order_status) {
      order->set_cancel_confirmed(true);
    } else {
      LOG(ERROR) << "[ParseFIXOrderInfo] Unknown order status! " << order_status;
    }
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

auto BitstampParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg.type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  if (msg.type == OrderSubTopicId::QUICKFIX_MESSAGE) {
    ParseFixMessage();
    return result_;
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      // pass
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpensInfo();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalance();
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  return result_;
}

const std::string& BitstampParser::GetLocaleSymbol(const std::string& symbol) {
  if (fix_native_map_.count(symbol) == 0) {
    auto locale_symbol = boost::algorithm::to_lower_copy(symbol);
    boost::replace_all(locale_symbol, "/", "");
    fix_native_map_.emplace(symbol, locale_symbol);
  }
  return fix_native_map_.at(symbol);
}

}  // namespace coin2::exchange::bitstamp::order_v2
